package com.pine.dcts.service;

import com.alibaba.datax.common.constant.PluginType;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.core.util.container.CoreConstant;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.pine.dcts.DctsEngine;
import com.pine.dcts.common.DctsException;
import com.pine.dcts.common.TaskStatusType;
import com.pine.dcts.config.DctsConfig;
import com.pine.dcts.core.IPluginConfigHandler;
import com.pine.dcts.core.impl.CoreConfigHandler;
import com.pine.dcts.entity.JobInfo;
import com.pine.dcts.model.JobVO;
import com.pine.dcts.taskScheduling.common.Constants;
import com.pine.dcts.taskScheduling.job.DataxJob;
import com.pine.dcts.taskScheduling.scheduler.DynamicScheduler;
import com.pine.dcts.utils.CopierUtil;
import lombok.extern.slf4j.Slf4j;
import org.quartz.SchedulerException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.Objects;

import static com.pine.dcts.common.DctsConstant.CONFIG_SCHEDUL;
import static com.pine.dcts.common.DctsConstant.CONFIG_SCHEDUL_EXETYPE;

@Service
@Slf4j
public class TaskService {

    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    @Autowired
    private JobService jobService;
    @Autowired
    private List<IPluginConfigHandler> pluginConfigHandlers;

    public void taskStart(Integer jobId) {
        JobVO jobVO = jobService.getJobVOById(jobId);
        JobInfo jobInfo  = new JobInfo();
        jobInfo.setId(jobId);
        jobInfo.setStatus(TaskStatusType.PREPARE.getValue());
        jobService.updateById(jobInfo);
        JSONObject jsonObject = jobVO.getConfig().getJSONObject(CONFIG_SCHEDUL);
//        if (Objects.equals(jsonObject.getInteger(CONFIG_SCHEDUL_EXETYPE), Constants.NO_SCHEDULER_TYPE)) {
//            jobInfo.setStatus(TaskStatusType.RUNNING.getValue());
//            jobService.updateById(jobInfo);
//            threadPoolTaskExecutor.execute(() -> {
//                Configuration configuration = Configuration.from(JobConfigBuilder.buildConfig(jobVO));
//                new CoreConfigHandler().supplementCoreConfig(configuration);
//                jobVO.getItems().forEach(item -> pluginConfigHandlers.stream()
//                        .filter(h -> h.support(item.getName())).findFirst().ifPresent(pluginConfigHandler -> pluginConfigHandler.supplementConfig(PluginType.valueOf(item.getType().toUpperCase())
//                        , configuration)));
//                DctsEngine engin = new DctsEngine();
//                engin.start(configuration.toJSON());
//                jobInfo.setStatus(TaskStatusType.FINISH.getValue());
//                jobService.updateById(jobInfo);
//            });
//        } else {
//            DataxJob dataxJob = JSON.toJavaObject(jsonObject, DataxJob.class);
//            dataxJob.setId(String.valueOf(jobVO.getId()));
//            dataxJob.setJobName(jobVO.getName());
//            try {
//                DynamicScheduler.addJob(dataxJob);
//               // Thread.sleep(3000);
//                DynamicScheduler.pauseJob(dataxJob);
//            } catch (SchedulerException e) {
//                e.printStackTrace();
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }

        DataxJob dataxJob = JSON.toJavaObject(jsonObject, DataxJob.class);
        dataxJob.setId(String.valueOf(jobVO.getId()));
        dataxJob.setJobName(jobVO.getName());
        try {
            DynamicScheduler.triggerJob(dataxJob);
             Thread.sleep(2000);
            DynamicScheduler.pauseJob(dataxJob);
        } catch (SchedulerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }

    @Async(value = "asyncTaskExecutor")
    public void taskStart1(Integer jobId) {
        log.info("当前线程名字是：{}", Thread.currentThread().getName());
        Configuration configuration = DctsConfig.jobConfigCenter.get(String.valueOf(jobId));
        if (configuration == null) {
            throw new DctsException("当前job配置不存在");
        }
        DctsEngine engin = new DctsEngine();
        engin.start(configuration.toJSON());
        log.info("当前线程名字是：{}", Thread.currentThread().getName());
    }
}
